package com.apps.sdses.flink141.demo;

import com.google.gson.Gson;
import org.apache.flink.table.functions.FunctionContext;
import org.apache.flink.table.functions.TableFunction;
import org.apache.flink.types.Row;

import java.util.HashMap;
import java.util.Map;
import java.util.Set;

/**
 * author: codejiwei
 * date: 2023/8/24
 * desc:
 **/
public class FieldsToJsonUDF extends TableFunction<Row> {
    Gson gson = null;

    @Override
    public void open(FunctionContext context) throws Exception {
        gson = new Gson();
        super.open(context);
    }

    public String eval(Row row, String[] fieldIndex) {
        Set<String> names = row.getFieldNames(true);
        Map<String, Object> map = new HashMap<>();

        for (String name : names) {
            map.put(name, row.getFieldAs(name));
        }
        return gson.toJson(map);
    }
}
